package project

import dmputils.{GeoCodingApi, PeiZhiFile}
import jdk.nashorn.internal.parser.JSONParser
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.parsing.json.JSONObject

object SQBiaoQian {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("商圈标签化")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc: SparkContext = new SparkContext(conf)
    val sqlc: SQLContext = new SQLContext(sc)
    import sqlc.implicits._


    val dataFrame: DataFrame = sqlc.read.parquet(PeiZhiFile.config.getString("inputpath"))

    val baseData: RDD[Row] = dataFrame.map(t => t).filter(f => !f.getAs[String]("long").isEmpty && !f.getAs[String]("lat").isEmpty)

    val latAndLng: RDD[String] = baseData.map(f => {
      val lng: String = f.getAs[String]("long")
      val lat: String = f.getAs[String]("lat")

      lat + "," + lng
    })

    val businesstagFromBaidu: RDD[String] = latAndLng.map(latAndLng => {

      val tagFromBaidu: String = GeoCodingApi.bussinessTagFromBaidu(latAndLng)

      tagFromBaidu


    })
    businesstagFromBaidu.collect.foreach(println)




    sc.stop()
  }
}
